package cn.assumejson.pro_netty.netty;

import cn.assumejson.pro_netty.constant.Constants;
import cn.assumejson.pro_netty.register.zk.ServerWatcher;
import cn.assumejson.pro_netty.register.zk.ZookeeperFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;

import java.net.InetAddress;

import static cn.assumejson.pro_netty.constant.Constants.port;
import static cn.assumejson.pro_netty.register.zk.ZookeeperFactory.SERVER_PATH_REGISTER;

/**
 * @author by gang.Li 2022/3/12 10:21
 */
public class NettyServer {

    private static final Logger log = org.slf4j.LoggerFactory.getLogger(NettyServer.class);

    public static void nettyStart() {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup(1);

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(workerGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            sc.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,
                                    0, 4, 0, 4));
                            sc.pipeline().addLast(new StringDecoder());
                            sc.pipeline().addLast(new ServerHandler());
                            sc.pipeline().addLast(new LengthFieldPrepender(4, false));
                            sc.pipeline().addLast(new StringEncoder());
                        }
                    });

            // bind port
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            log.info("Netty service started successfully....");
            // TODO 连接注册中心 获取相应的配置（这里需要使用SPI机制来获取支持的注册中心）
            CuratorFramework client = ZookeeperFactory.create();
            // 获取当前服务的IP
            InetAddress localHost = InetAddress.getLocalHost();
            Stat stat = client.checkExists().forPath(SERVER_PATH_REGISTER);
            if (stat == null) {
                client.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT).forPath(SERVER_PATH_REGISTER, "0".getBytes());
            }
            //在SERVER_PATH目录下构建临时节点，如10.118.15.15#8080#1#0000000000
            client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
                    forPath(SERVER_PATH_REGISTER + "/" + localHost.getHostAddress() + "#" + port + "#" + Constants.weight + "#");
            //服务端需要需要加上zk的监控，以防Session断了，导致临时节点丢失
            ServerWatcher.serverKey = localHost.getHostAddress() + port + Constants.weight;
            client.getChildren().usingWatcher(ServerWatcher.getInstance()).forPath(SERVER_PATH_REGISTER);

            log.info("netty server is running .....");
            // 阻塞主线程，直到socket通道被关闭
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void start() {
        nettyStart();
    }
}
